package com.bbx.flink.demo.data_stream_state.functions;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;

import java.util.ArrayList;
import java.util.List;

@Slf4j
public class OperatorStateDemoFunction implements MapFunction<String, Integer> , CheckpointedFunction {
    private static final long serialVersionUID = -7520961398558708631L;

    private  Integer  sum = 0;

    private  transient  ListState<Integer> checkpointedState ;


    /**
     * 用来存放元素
     */
    private List<Integer> valueList = new ArrayList<Integer>();

    @Override
    public Integer map(String value) throws Exception {
        //数据进来后 计数器加1
        valueList.add(Integer.parseInt(value));
        ++sum;
        return sum;
    }

    /**
     *
     * @param context
     * @throws Exception
     */
    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        log.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~[{}]","snapshotState");
        checkpointedState.clear();
        for (Integer integer : valueList) {
           checkpointedState.add(integer);
        }


    }

    /**
     * 第一次创建时会被调用
     * 恢复时也会被调用
     * @param context
     * @throws Exception
     */
    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        log.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~[{}]","initializeState");
        checkpointedState = context.getOperatorStateStore().getListState(new ListStateDescriptor<Integer>(
                "operatorState", Integer.class));
        if (context.isRestored()) {
            for(Integer i :checkpointedState.get()){
                valueList.add(i);
            }
        }
    }


}
